/*
 * Copyright 1999-2018 Alibaba Group Holding Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.nacos.config.server.service;

import static com.alibaba.nacos.config.server.utils.LogUtil.memoryLog;
import static com.alibaba.nacos.config.server.utils.LogUtil.pullLog;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import javax.servlet.AsyncContext;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Service;

import com.alibaba.nacos.config.server.model.SampleResult;
import com.alibaba.nacos.config.server.utils.GroupKey;
import com.alibaba.nacos.config.server.utils.LogUtil;
import com.alibaba.nacos.config.server.utils.MD5Util;
import com.alibaba.nacos.config.server.utils.RequestUtil;
import com.alibaba.nacos.config.server.utils.event.EventDispatcher.AbstractEventListener;
import com.alibaba.nacos.config.server.utils.event.EventDispatcher.Event;

/**
 * 长轮询服务。负责处理
 * @author Nacos
 */
@Service
public class LongPullingService extends AbstractEventListener {

	private static final int FIXED_POLLING_INTERVAL_MS = 10000;
    
    private static final int SAMPLE_PERIOD = 100;
    
    private static final int SAMPLE_TIMES = 3;
    
    private static final String TRUE_STR = "true";

	private Map<String, Long> retainIps = new ConcurrentHashMap<String, Long>();
    
    private static boolean isFixedPolling() {
        return SwitchService.getSwitchBoolean(SwitchService.FIXED_POLLING, false);
    }

    private static int getFixedPollingInterval() {
        return SwitchService.getSwitchInteger(SwitchService.FIXED_POLLING_INTERVAL, FIXED_POLLING_INTERVAL_MS);
    }


    public boolean isClientLongPolling(String clientIp) {
        return getClientPollingRecord(clientIp) != null;
    }

    public Map<String, String> getClientSubConfigInfo(String clientIp) {
        ClientLongPulling record = getClientPollingRecord(clientIp);

        if (record == null) {
        	return Collections.<String, String>emptyMap();
        }

        return record.clientMd5Map;
    }

	public SampleResult getSubscribleInfo(String dataId, String group, String tenant) {
		String groupKey = GroupKey.getKeyTenant(dataId, group, tenant);
		SampleResult sampleResult = new SampleResult();
		Map<String, String> lisentersGroupkeyStatus = new HashMap<String, String>(50);

		for (ClientLongPulling clientLongPulling : allSubs) {
			if (clientLongPulling.clientMd5Map.containsKey(groupKey)) {
				lisentersGroupkeyStatus.put(clientLongPulling.ip, clientLongPulling.clientMd5Map.get(groupKey));
			}
		}
		sampleResult.setLisentersGroupkeyStatus(lisentersGroupkeyStatus);
		return sampleResult;
	}
	
	public SampleResult getSubscribleInfoByIp(String clientIp) {
		SampleResult sampleResult = new SampleResult();
		Map<String, String> lisentersGroupkeyStatus = new HashMap<String, String>(50);
		
		for (ClientLongPulling clientLongPulling : allSubs) {
			if (clientLongPulling.ip.equals(clientIp)) {
				// 一个ip可能有多个监听
				if (!lisentersGroupkeyStatus.equals(clientLongPulling.clientMd5Map)) {
					lisentersGroupkeyStatus.putAll(clientLongPulling.clientMd5Map);
				}
			}
		}
		sampleResult.setLisentersGroupkeyStatus(lisentersGroupkeyStatus);
		return sampleResult;
	}
	
	/**
	 * 聚合采样结果中的采样ip和监听配置的信息；合并策略用后面的覆盖前面的是没有问题的
	 * @param sampleResults sample Results
	 * @return Results
	 */
	public SampleResult mergeSampleResult(List<SampleResult> sampleResults) {
		SampleResult mergeResult = new SampleResult();
		Map<String, String> lisentersGroupkeyStatus = new HashMap<String, String>(50);
		for (SampleResult sampleResult : sampleResults) {
			Map<String, String> lisentersGroupkeyStatusTmp = sampleResult.getLisentersGroupkeyStatus();
			for (Map.Entry<String, String> entry : lisentersGroupkeyStatusTmp.entrySet()) {
				lisentersGroupkeyStatus.put(entry.getKey(), entry.getValue());
			}
		}
		mergeResult.setLisentersGroupkeyStatus(lisentersGroupkeyStatus);
		return mergeResult;
	}
	
    public Map<String, Set<String>> collectApplicationSubscribeConfigInfos() {
    	if (allSubs == null || allSubs.isEmpty()) {
			return null;
		}
    	HashMap<String, Set<String>> app2Groupkeys = new HashMap<String, Set<String>>(50);
    	for (ClientLongPulling clientLongPulling : allSubs) {
    		if(StringUtils.isEmpty(clientLongPulling.appName) || "unknown".equalsIgnoreCase(clientLongPulling.appName)) {
    			continue;
    		}
    		Set<String> appSubscribeConfigs = app2Groupkeys.get(clientLongPulling.appName);
    		Set<String> clientSubscribeConfigs = clientLongPulling.clientMd5Map.keySet();
    		if(appSubscribeConfigs==null) {
    			appSubscribeConfigs = new HashSet<String>(clientSubscribeConfigs.size());
    		} 
    	    appSubscribeConfigs.addAll(clientSubscribeConfigs);
    		app2Groupkeys.put(clientLongPulling.appName, appSubscribeConfigs);
    	}
    	
    	return app2Groupkeys;
    }

	
	public SampleResult getCollectSubscribleInfo(String dataId, String group, String tenant) {
		List<SampleResult> sampleResultLst = new ArrayList<SampleResult>(50);
		for (int i = 0; i < SAMPLE_TIMES; i++) {
			SampleResult sampleTmp = getSubscribleInfo(dataId, group, tenant);
			if (sampleTmp != null) {
				sampleResultLst.add(sampleTmp);
			}
			if (i < SAMPLE_TIMES - 1) {
				try {
					Thread.sleep(SAMPLE_PERIOD);
				} catch (InterruptedException e) {
					LogUtil.clientLog.error("sleep wrong", e);
				}
			}
		}
		
		SampleResult sampleResult = mergeSampleResult(sampleResultLst);
		return sampleResult;
	}
	
	public SampleResult getCollectSubscribleInfoByIp(String ip) {
		SampleResult sampleResult = new SampleResult();
		sampleResult.setLisentersGroupkeyStatus(new HashMap<String, String>(50));
		for (int i = 0; i < SAMPLE_TIMES; i++) {
			SampleResult sampleTmp = getSubscribleInfoByIp(ip);
			if (sampleTmp != null) {
				if (sampleTmp.getLisentersGroupkeyStatus() != null
						&& !sampleResult.getLisentersGroupkeyStatus().equals(sampleTmp.getLisentersGroupkeyStatus())) {
					sampleResult.getLisentersGroupkeyStatus().putAll(sampleTmp.getLisentersGroupkeyStatus());
				}
			}
			if (i < SAMPLE_TIMES - 1) {
				try {
					Thread.sleep(SAMPLE_PERIOD);
				} catch (InterruptedException e) {
					LogUtil.clientLog.error("sleep wrong", e);
				}
			}
		}
		return sampleResult;
	}
    
    private ClientLongPulling getClientPollingRecord(String clientIp) {
        if (allSubs == null) {
        	return null;
        }

        for (ClientLongPulling clientLongPulling : allSubs) {
            HttpServletRequest request = (HttpServletRequest) clientLongPulling.asyncContext.getRequest();

            if (clientIp.equals(RequestUtil.getRemoteIp(request))) {
                return clientLongPulling;
            }
        }

        return null;
    }

    public void addLongPullingClient(HttpServletRequest req, HttpServletResponse rsp, Map<String, String> clientMd5Map, int probeRequestSize) {

        String str = req.getHeader(LongPullingService.LONG_PULLING_HEADER);
        String noHangUpFlag = req.getHeader(LongPullingService.LONG_PULLING_NO_HANG_UP_HEADER);
        String appName = req.getHeader(RequestUtil.CLIENT_APPNAME_HEADER);
		String tag = req.getHeader("Vipserver-Tag");
        int delayTime=SwitchService.getSwitchInteger(SwitchService.FIXED_DELAY_TIME, 500);
        /**
         * 提前500ms返回响应，为避免客户端超时 @qiaoyi.dingqy 2013.10.22改动  add delay time for LoadBalance
         */
        long timeout = Math.max(10000, Long.parseLong(str) - delayTime); 
        if (isFixedPolling()) {
            timeout = Math.max(10000, getFixedPollingInterval());
            // do nothing but set fix polling timeout
        } else {
            long start = System.currentTimeMillis();
            List<String> changedGroups = MD5Util.compareMd5(req, rsp, clientMd5Map);
            if (changedGroups.size() > 0) {
                generateResponse(req, rsp, changedGroups);
                LogUtil.clientLog.info("{}|{}|{}|{}|{}|{}|{}",
                        new Object[]{System.currentTimeMillis() - start, "instant", RequestUtil.getRemoteIp(req), "polling",
                                clientMd5Map.size(), probeRequestSize, changedGroups.size()});
                return;
            } else if(noHangUpFlag!=null && noHangUpFlag.equalsIgnoreCase(TRUE_STR)) {
            	LogUtil.clientLog.info("{}|{}|{}|{}|{}|{}|{}",
                        new Object[]{System.currentTimeMillis() - start, "nohangup", RequestUtil.getRemoteIp(req), "polling",
                                clientMd5Map.size(), probeRequestSize, changedGroups.size()}); 
            	return;
            }
        }
        String ip = RequestUtil.getRemoteIp(req);
        // 一定要由HTTP线程调用，否则离开后容器会立即发送响应
        final AsyncContext asyncContext = req.startAsync();
        // AsyncContext.setTimeout()的超时时间不准，所以只能自己控制
        asyncContext.setTimeout(0L);
        

		scheduler.execute(
				new ClientLongPulling(asyncContext, clientMd5Map, ip, probeRequestSize, timeout, appName, tag));
    }

    @Override
    public List<Class<? extends Event>> interest() {
        List<Class<? extends Event>> eventTypes = new ArrayList<Class<? extends Event>>();
        eventTypes.add(LocalDataChangeEvent.class);
        return eventTypes;
    }

    @Override
    public void onEvent(Event event) {
        if (isFixedPolling()) {
            // ignore
        } else {
			if (event instanceof LocalDataChangeEvent) {
				LocalDataChangeEvent evt = (LocalDataChangeEvent) event;
				scheduler.execute(new DataChangeTask(evt.groupKey, evt.isBeta, evt.betaIps));
			}
        }
    }

    static public boolean isSupportLongPulling(HttpServletRequest req) {
        return null != req.getHeader(LONG_PULLING_HEADER);
    }

    @SuppressWarnings("PMD.ThreadPoolCreationRule")
    public LongPullingService() {
        allSubs = new ConcurrentLinkedQueue<ClientLongPulling>();

        scheduler = Executors.newScheduledThreadPool(1, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setDaemon(true);
                t.setName("com.alibaba.nacos.LongPulling");
                return t;
            }
        });
        scheduler.scheduleWithFixedDelay(new StatTask(), 0L, 10L, TimeUnit.SECONDS);
    }

    // =================

    static public final String LONG_PULLING_HEADER = "Long-Pulling-Timeout";
    static public final String LONG_PULLING_NO_HANG_UP_HEADER = "Long-Pulling-Timeout-No-Hangup";

    final ScheduledExecutorService scheduler;

	/**
	 * 长轮询订阅关系
	 */
	final Queue<ClientLongPulling> allSubs;

    // =================

    class DataChangeTask implements Runnable {
        @Override
        public void run() {
        	try {
        		ConfigService.getContentBetaMd5(groupKey);
        		for (Iterator<ClientLongPulling> iter = allSubs.iterator(); iter.hasNext(); ) {
                    ClientLongPulling clientSub = iter.next();
                    if (clientSub.clientMd5Map.containsKey(groupKey)) {
                    	// 如果beta发布且不在beta列表直接跳过
                    	if (isBeta && !betaIps.contains(clientSub.ip)) {
							continue;
						}
                    	
						// 如果tag发布且不在tag列表直接跳过
						if (StringUtils.isNotBlank(tag) && !tag.equals(clientSub.tag)) {
							continue;
						}
						
                    	getRetainIps().put(clientSub.ip, System.currentTimeMillis());
                    	iter.remove(); // 删除订阅关系
                        LogUtil.clientLog.info("{}|{}|{}|{}|{}|{}|{}",
                                new Object[]{(System.currentTimeMillis() - changeTime),
                                        "in-advance", RequestUtil.getRemoteIp((HttpServletRequest) clientSub.asyncContext.getRequest()), "polling",
                                        clientSub.clientMd5Map.size(), clientSub.probeRequestSize, groupKey});
                        clientSub.sendResponse(Arrays.asList(groupKey));
                    }
                }
			} catch (Throwable t) {
				LogUtil.defaultLog.error("data change error:" + t.getMessage(), t.getCause());
			}
        }

		DataChangeTask(String groupKey) {
			this(groupKey, false, null);
		}

		DataChangeTask(String groupKey, boolean isBeta, List<String> betaIps) {
			this(groupKey, isBeta, betaIps, null);
		}

		DataChangeTask(String groupKey, boolean isBeta, List<String> betaIps, String tag) {
			this.groupKey = groupKey;
			this.isBeta = isBeta;
			this.betaIps = betaIps;
			this.tag = tag;
		}
        
        final String groupKey;
        final long changeTime = System.currentTimeMillis();
        final boolean isBeta;
        final List<String> betaIps;
        final String tag;
    }

    // =================

    class StatTask implements Runnable {
        @Override
        public void run() {
            memoryLog.info("[long-pulling] client count " + allSubs.size());
        }
    }

    // =================

    class ClientLongPulling implements Runnable {

        @Override
        public void run() {
            asyncTimeoutFuture = scheduler.schedule(new Runnable() {
                public void run() {
                	try {
                		getRetainIps().put(ClientLongPulling.this.ip, System.currentTimeMillis());
						/**
						 * 删除订阅关系
						 */
                		allSubs.remove(ClientLongPulling.this);

                        if(isFixedPolling()) {
                        	LogUtil.clientLog.info("{}|{}|{}|{}|{}|{}",
                        			new Object[]{(System.currentTimeMillis() - createTime),
                        			"fix", RequestUtil.getRemoteIp((HttpServletRequest) asyncContext.getRequest()), "polling",
                        			clientMd5Map.size(), probeRequestSize});
                            List<String> changedGroups = MD5Util.compareMd5((HttpServletRequest) asyncContext.getRequest(), (HttpServletResponse) asyncContext.getResponse(), clientMd5Map);
                            if (changedGroups.size() > 0) {
                                sendResponse(changedGroups);
                            } else {
                                sendResponse(null);
                            }
                        } else {
                        	LogUtil.clientLog.info("{}|{}|{}|{}|{}|{}",
                        			new Object[]{(System.currentTimeMillis() - createTime),
                        			"timeout", RequestUtil.getRemoteIp((HttpServletRequest) asyncContext.getRequest()), "polling",
                        			clientMd5Map.size(), probeRequestSize});
                            sendResponse(null);
                        }
        			} catch (Throwable t) {
        				LogUtil.defaultLog.error("long pulling error:" + t.getMessage(), t.getCause());
        			}
                    
                }
            }, timeoutTime, TimeUnit.MILLISECONDS);

            allSubs.add(this);
        }

        void sendResponse(List<String> changedGroups) {
        	/**
        	 *  取消超时任务
        	 */
        	if (null != asyncTimeoutFuture) { 
                asyncTimeoutFuture.cancel(false);
            }
            generateResponse(changedGroups);
        }

        void generateResponse(List<String> changedGroups) {
            if (null == changedGroups) {
				/**
				 * 告诉容器发送HTTP响应
				 */
            	asyncContext.complete();
                return;
            }

            HttpServletResponse response = (HttpServletResponse) asyncContext.getResponse();

            try {
                String respString = MD5Util.compareMd5ResultString(changedGroups);

                // 禁用缓存
                response.setHeader("Pragma", "no-cache");
                response.setDateHeader("Expires", 0);
                response.setHeader("Cache-Control", "no-cache,no-store");
                response.setStatus(HttpServletResponse.SC_OK);
                response.getWriter().println(respString);
                asyncContext.complete(); 
            } catch (Exception se) {
                pullLog.error(se.toString(), se);
                asyncContext.complete(); 
            }
        }
        
		ClientLongPulling(AsyncContext ac, Map<String, String> clientMd5Map, String ip, int probeRequestSize,
				long timeoutTime, String appName, String tag) {
			this.asyncContext = ac;
			this.clientMd5Map = clientMd5Map;
			this.probeRequestSize = probeRequestSize;
			this.createTime = System.currentTimeMillis();
			this.ip = ip;
			this.timeoutTime = timeoutTime;
			this.appName = appName;
			this.tag = tag;
		}

        // =================

        final AsyncContext asyncContext;
        final Map<String, String> clientMd5Map;
        final long createTime;
        final String ip;
        final String appName;
        final String tag;
        final int probeRequestSize;
        final long timeoutTime;

        Future<?> asyncTimeoutFuture;
    }

    void generateResponse(HttpServletRequest request, HttpServletResponse response, List<String> changedGroups) {
        if (null == changedGroups) {
            return;
        }

        try {
            String respString = MD5Util.compareMd5ResultString(changedGroups);
            // 禁用缓存
            response.setHeader("Pragma", "no-cache");
            response.setDateHeader("Expires", 0);
            response.setHeader("Cache-Control", "no-cache,no-store");
            response.setStatus(HttpServletResponse.SC_OK);
            response.getWriter().println(respString);
        } catch (Exception se) {
            pullLog.error(se.toString(), se);
        }
    }

	public Map<String, Long> getRetainIps() {
		return retainIps;
	}

	public void setRetainIps(Map<String, Long> retainIps) {
		this.retainIps = retainIps;
	}


}
